
#include "manager.h"

Err rp_rstatd_manager()
{
	char fname[] = "rp_rstatd_manager_run";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(&g_mutex_rstatd_manager);

	struct event_base *evbase;
	struct event *ev, *ev2;
	
	evbase = event_init();
	evtimer_set(ev, rp_rstatd_manager_monitor, NULL);
	if (evtimer_add(ev, &g_timeval_rstatd_manager) < 0)
	{
		rp_log(LOG_FATAL, fname, msg);
		return err;
	}
	event_loop(EVLOOP_NONBLOCK);

	err = rp_mutex_unlock(&g_mutex_rstatd_manager);
	return err;	
}

void rp_rstatd_manager_monitor(int fd, short event, void *arg) 
{
	char fname[] = "rp_rstatd_manager_run";
	char msg[] = "fatal error";
	Err err;
	Snode *snode, *temp;
	
	/*	cleanup the going-away rpointd   */
	err = rp_ptable_cleanup();
	
	/*	finish the previous command   */
	switch (g_state_rstatd)
	{
		case (ST_EXIT) :	{
			snode = rp_rstatd_scoreboard_remove_idle(&g_scoreboard_rstatd, 0);
			while (snode) 
			{
				temp = snode;
				temp->score->state = ST_EXIT;
				snode = snode->next;
				err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)temp);
			}	
		
			if (!rp_rstatd_scoreboard_total(&g_scoreboard_rstatd))
			{		
				// exit event dispatch
				//event_loopbreak();
				struct timeval time;
				time.tv_sec = 0;
				time.tv_usec = 0;
				event_loopexit(&time);
			}		
			return;
		}
		case (ST_PAUSE) : {
			
			return;
		}
		default : {}
	}
	
	/*	handle the next pending command   */
	Cmd* cmd = rp_cqueue_get(&g_cqueue_rstatd_cmd);
 	if (!cmd)
	{
		rp_log(LOG_DEBUG, fname, "No pending commands");
	} else {
		rp_log(LOG_DEBUG, fname, "Processing the command");	
		
		switch (cmd->code)
		{
			case (OP_REPLICA): {
				// TODO: we need some mechanism to check rpcopy status
				rp_pid_t pid = rp_process_create("rpcopy",cmd->arg, NULL);
				if (pid<0) rp_log(LOG_CRITICAL, fname, "Failed to execute rpcopy");
				break;
			}
			case (OP_SHUTDOWN): {
				/*	mark the exit state & save the snode onto ptable   */
				g_state_rstatd = ST_EXIT;
				snode = rp_rstatd_scoreboard_remove_idle(&g_scoreboard_rstatd, 0);  //peng
				while (snode) 
				{
					temp = snode;
					temp->score->state = ST_EXIT;
					snode = snode->next;
					err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)temp);
				}			
				return;
			}
			case (OP_SHUTDOWN_FORCE): {
				/*	mark the exit state & save the snode onto ptable   */
				g_state_rstatd = ST_EXIT;
				snode = rp_rstatd_scoreboard_remove_total(&g_scoreboard_rstatd);   //peng
				while (snode) 
				{
					temp = snode;
					temp->score->state = ST_EXIT;
					snode = snode->next;
					err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)temp);
				}			
				return;
			}
			case (OP_PAUSE): {

				return;
			}
			case (OP_RESUME): {

				break;
			}
			default: {
				rp_log(LOG_WARNING, fname, "Invalid command");
				break;	
			}
		}
		free(cmd);
	}

	/*	manage rpointd process	    */	
	int extra;
	char fpath[MAX_FILENAME_LENGTH];	
	err = rp_rstatd_scoreboard_monitor(&g_scoreboard_rstatd, &extra);

	if (extra < 0)
	{	/*	mark the exit state & save the snode onto ptable   */
	  snode = rp_rstatd_scoreboard_remove_idle(&g_scoreboard_rstatd, abs(extra));  //peng
		while (!snode) 
		{
			temp = snode;
			snode = snode->next;
			temp->prev = NULL;
			temp->next = NULL;
			if (!snode) snode->prev = NULL;
			temp->score->state = ST_EXIT;
			err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)temp);						
		}
	} else if (extra > 0) 
	{	/*	create new rpointd		*/
	        int i;
	        for (i=0; i<abs(extra); i++)
		{		
			snode = (Snode*)malloc(sizeof(Snode));
			
			if ((snode->pid = rp_process_create("rpointd", NULL)) <= 0)
			{
				free(snode);
				rp_log(LOG_CRITICAL, fname, "Fail to create new rpointd process");
				continue;
			} 
		
			sprintf(fpath, "/tmp/rally/%d", snode->pid);
			if ((snode->fd = open(fpath, O_CREAT)) < 0)
			{
				err = rp_process_terminate(snode->pid, SIGTERM);		// Forceful kill the new rpointd process
				/*	pid 	= xxx
			 	 *	fd   	= 0
			 	 *	state= 0
		 		 */
				err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)snode);
				rp_log(LOG_CRITICAL, fname, "Cannot open file under /tmp/rally");
				continue;
			} 
			
			if (mmap(&snode->score, sizeof(Score), PROT_READ|PROT_WRITE, 
			          MAP_SHARED|MAP_LOCKED, fd, 0) == MAP_FAILED)
			{
				close(snode->fd);
				err = rp_process_terminate(snode->pid, SIGTERM);		// Forceful kill the new rpointd process
				/*	pid 	= xxx
			 	 *	fd   	= xxx
			 	 *	state= 0
		 		 */
				err = rp_ptable_put(&g_ptable_rstatd, (Pnode*)snode);
				rp_log(LOG_CRITICAL, fname, "Failed mmap file under /tmp/rally");
				continue;
			} 
			/*	pid 	= xxx
		 	*	fd   	= xxx
		 	*	state= SS_IDLE
		 	*/
			 snode->score->state = SS_IDLE;
			err = rp_rstatd_scoreboard_append(&g_scoreboard_rstatd, snode);	
		}
	}
	
	return;			
}

